package com.huonan.utils;

import java.io.File;
import java.io.IOException;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * 各种操作hdfs的API
 */
public class HDFSUtil {
    private static final Logger logger = LogManager.getLogger("HDFSUtil");

    private static Configuration conf;
    static FileSystem fs = null;


    public static void uploadLocalFile2HDFS(String s, String d)
            throws IOException {
        FileSystem hdfs = FileSystem.get(getConf());
        Path src = new Path(s);
        Path dst = new Path(d);
        hdfs.copyFromLocalFile(src, dst);
        hdfs.close();
    }

    public static boolean hdfsPathexists(String path) throws IOException {
        return FileSystem.get(getConf()).exists(new Path(path));
    }


    public static Configuration getConf() {
        if (null == conf) {
            System.setProperty("user.name", "quantum_social");
            System.setProperty("HADOOP_USER_NAME", "quantum_social");
            conf = new Configuration();
        }
        return conf;
    }

    /**
     * @param
     * @return FileSystem
     * @Description get configured filesystem implementation
     **/
    public static FileSystem getFs(String user) {

        Configuration conf = new Configuration();
        conf.addResource("viewfs-mount-table.xml");
        conf.addResource("core-site.xml");
        conf.addResource("hdfs-site.xml");
        System.setProperty("user.name", "quantum_social");
        System.setProperty("HADOOP_USER_NAME", "quantum_social");
        //conf.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        try {
            fs = FileSystem.get(conf);
        } catch (IOException e) {
            logger.error(e.getMessage());
        }
        return fs;
    }


    /**
     * @param sorceFile
     * @param targetFile
     * @return void
     * @Description 上传本地file到hdfs，删除src
     **/
    public static int putLocalFileToHdfsDelSrc(String sorceFile, String targetFile, FileSystem fs, boolean delSrc) {


        FsPermission filePermission = new FsPermission(
                FsAction.ALL,
                FsAction.ALL,
                FsAction.ALL);
        Path srcPath = new Path(sorceFile);
        Path dstPath = new Path(targetFile);
        try {
            if (!fs.exists(dstPath)) {
                fs.mkdirs(dstPath, filePermission);
            }
            fs.copyFromLocalFile(delSrc, srcPath, dstPath);
        } catch (Exception e) {
            e.printStackTrace();
            return -1;
        }
        return 0;
    }

    /**
     * @param sourceFile
     * @param descFile
     * @return void
     * @Description 下载hdfs文件到本地
     **/
    public static void downHdfsFileToLocal(String sourceFile, String descFile) throws Exception {

        FileSystem fs = getFs("quantum_social");
        Path srcPath = new Path(sourceFile);
        Path dstPath = new Path(descFile);
        File localFile = new File(descFile.substring(0, descFile.lastIndexOf("/")));
        try {
            if (!localFile.exists()) {
                localFile.mkdirs();
                localFile.setReadable(true);
                localFile.setWritable(true);
                localFile.setExecutable(true);
            }
            fs.copyToLocalFile(srcPath, dstPath);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            HDFSUtil.closeFs(fs);
        }

    }

    public static void mkdirLocalFile(String fileName) throws IOException {

        FsPermission filePermission = new FsPermission(
                FsAction.ALL,
                FsAction.ALL,
                FsAction.ALL);
        FileSystem fs = HDFSUtil.getFs("");
        Path dstPath = new Path(fileName);
        if (!fs.exists(dstPath)) {
            fs.mkdirs(dstPath);
        }
    }

    public static void closeFs(FileSystem fs) throws IOException {
        if (fs != null) {
            fs.close();
        }
    }

    /**
     * @param sourceFileName
     * @param zipName
     * @return void
     * @Description 压缩文件(指定压缩格式)
     **/
    public static void compress(String sourceFileName, String zipName, String compressType) {
        //compress("org.apache.hadoop.io.compress.GzipCodec")
        System.out.printf("Compress : [source:%s] [target:%s] \n", sourceFileName, zipName);
        Class<?> codecClass = null;
        FSDataInputStream in = null;
        CompressionOutputStream out = null;
        try {
            //String name = "org.apache.hadoop.io.co";
            //org.apache.hadoop.io.compress.Compressor
            codecClass = Class.forName("java.util.zip.DeflaterOutputStream");
            FileSystem fs = getFs("");

            CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass,
                    fs.getConf());
            //指定压缩文件路径
            //FSDataOutputStream outputStream = fs.create(new Path("/user/hadoop/text.gz"));
            FSDataOutputStream outputStream = fs.create(new Path(zipName));
            //指定要被压缩的文件路径
            // FSDataInputStream in = fs.open(new Path("/user/hadoop/aa.txt"));
            in = fs.open(new Path(sourceFileName));
            //创建压缩输出流
            out = codec.createOutputStream(outputStream);
            org.apache.hadoop.io.IOUtils.copyBytes(in, out, conf);

            System.out.printf("【Compress success**】");
            System.out.printf("");
        } catch (Exception e) {
            System.out.printf("【Compress failed】： [source]" +
                    ":{} [target]:{} \n", sourceFileName, zipName);
        } finally {
            org.apache.hadoop.io.IOUtils.closeStream(in);
            org.apache.hadoop.io.IOUtils.closeStream(out);
        }

    }

    /**
     * @param filePath
     * @return void
     * @Description 删除文件
     **/
    public static void deleteFile(String filePath) {
        FileSystem fs = getFs("");
        try {
            fs.delete(new Path(filePath), true);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void putMoveHdfsFile(String tmpZipFile, String DestHdfsFile, FileSystem fs) {
        FsPermission filePermission = new FsPermission(
                FsAction.ALL,
                FsAction.ALL,
                FsAction.ALL);
        Path srcPath = new Path(tmpZipFile);
        Path destDath = new Path(DestHdfsFile);
        Path dstDir = new Path(DestHdfsFile.substring(0, DestHdfsFile.lastIndexOf("/")));
        try {
            if (!fs.exists(srcPath)) {
                return;
            }
            if (!fs.exists(dstDir)) {
                fs.mkdirs(dstDir, filePermission);
            }
            fs.rename(srcPath, destDath);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * @return java.util.List<java.lang.String>
     * @Description 获取指定路径下的文件
     * @Param [path]
     **/
    public static FileStatus[] listHdfsFile(String path) {

        FileSystem fs = getFs("quantum_social");
        TreeMap<Long, String> mapList = new TreeMap<Long, String>();
        FileStatus[] tlistFiles = new FileStatus[0];
        try {
            if (fs.exists(new Path(path))) {
                try {
                    tlistFiles = fs.listStatus(new Path(path));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        //return Arrays.stream(tlistFiles).map(hf -> hf.getPath().toString()).collect(Collectors.toList());
        return tlistFiles;
    }
}